package industry_2024.industry_10.indicator

import org.apache.hudi.DataSourceWriteOptions.{PARTITIONPATH_FIELD, PRECOMBINE_FIELD, RECORDKEY_FIELD}
import org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, date_format}

object indicator02 {
  def main(args: Array[String]): Unit = {
    /*
          编写scala代码，使用Spark根据hudi_gy_dwd层的fact_machine_data表统计出每日每台设备，状态为“运行”的时长（若运行无结束时间，则需根据
          时间判断这个设备的运行状态的下一个状态是哪条数据，将下一个状态的数据的时间置为这个设备运行状态的结束时间,如果设备数据的运行状态不存在下一个
          状态，则该设备这个阶段数据的运行状态不参与计算，即该设备的这个阶段数据的运行状态时长按0计算），将结果数据写入hudi_gy_dws层的
          表machine_data_total_time中，然后使用spark-sql的cli根据machine_id降序和machine_record_date升序排序查询前5条数据，将SQL语句
          复制粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中
          对应的任务序号下
     */
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("指标计算")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      /*
          spark.sql.legacy.parquet.datetimeRebaseModeInRead: 这个参数控制 Spark 在读取 Parquet 文件时如何处理日期时间数据的旧版和
          新版兼容性问题。LEGACY 模式使 Spark 按照较旧的方式解析日期时间数据，确保与早期版本生成的 Parquet 文件兼容。
      spark.sql.legacy.avro.datetimeRebaseModeInWrite: 这个参数控制 Spark 在写入 Avro 文件时如何处理日期时间数据的旧版和新版兼容性
      问题。LEGACY 模式使 Spark 按照旧版方式写入日期时间数据，以确保与早期版本的 Avro 文件兼容。
       */
      .config("spark.sql.legacy.parquet.datetimeRebaseModeInRead", "LEGACY")
      .config("spark.sql.legacy.avro.datetimeRebaseModeInWrite", "LEGACY")
      .enableHiveSupport()
      .getOrCreate()

//
//    spark.sql("show partitions hudi_gy_dwd10.fact_machine_data").show
//
//    spark.table("hudi_gy_dwd10.fact_machine_data").limit(5).show

    val fact_machine_data="hdfs://192.168.40.110:9000/user/hive/warehouse/hudi_gy_dwd10.db/fact_machine_data"
    spark.read.format("hudi").load(fact_machine_data)
      .where(col("etldate")==="20240906")
      .createOrReplaceTempView("fact_machine_data")

    spark.sql("select * from fact_machine_data").limit(5).show



    /*
        lead("字段",1,"1970-01-01") over():根据字段往后位移1行，如果没有下一行的数据，就赋值"1970-01-01"，至于为什么要赋值这个是因为根据题目
        需求,1970-01-01是时间戳的开始时间，所以换算成时间戳就是0，然后下面那一行的逻辑就是根据下一次运行的记录时间减去当前行的记录时间，得到两次记
        录的时间差


     */
val result=spark.sql(
  """
    |select
    |r2.machine_id,
    |sum(r2.time) as total_time,
    |r2.machine_record_date
    |from(
    |select
    |r1.machine_id,
    |r1.machine_record_date,
    |case
    |when r1.machinerecordstate="运行"
    |then unix_timestamp( lead(r1.machinerecorddate,1,"1970-01-01") over(partition by r1.machine_id order by r1.machinerecorddate) ) - unix_timestamp(r1.machinerecorddate)
    |else 0
    |end as time
    |from(
    |select distinct
    |f.machineid as machine_id,
    |substr(f.machinerecorddate,1,10) as machine_record_date,
    |f.machinerecordstate,
    |f.machinerecorddate
    |from fact_machine_data as f
    |) as r1
    |) as r2
    |group by r2.machine_id, r2.machine_record_date
    |""".stripMargin)

   val result_path="hdfs://192.168.40.110:9000/user/hive/warehouse/hudi_gy_dws10.db/machine_data_total_time"

    //  创建表(数据库要在hive的客户端里面创建，这里创建的好像不行)
    spark.sql("drop table if exists hudi_gy_dws10.machine_data_total_time")
    spark.sql(
      """
        |create table if not exists hudi_gy_dws10.machine_data_total_time(
        |machine_id int,
        |total_time int
        |)using hudi
        |tblproperties(
        |type="cow",
        |primaryKey="machine_id",
        |preCombineField="total_time",
        |hoodie.datasource.hive_aync.mode="hms"
        |)
        |partitioned by(machine_record_date String)
        |""".stripMargin)

    result.withColumn("machine_record_date",date_format(col("machine_record_date"),"yyyy-MM-dd"))
      .write.mode("append")
      .format("hudi")
      .options(getQuickstartWriteConfigs)
      .option(RECORDKEY_FIELD.key(),"machine_id")
      .option(PRECOMBINE_FIELD.key(),"total_time")
      .option(PARTITIONPATH_FIELD.key(),"machine_record_date")
      .option("hoodie.table.name","machine_data_total_time")
      .save(result_path)

    println("完成")



    spark.close()
  }

}
